package io.sqrtqiezi.practise.hadoop.sort.secondary;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.IOException;

public class SecondarySortMapper extends Mapper<LongWritable, Text, DateTemperaturePair, IntWritable> {
    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        final String line = value.toString();
        final String[] tokens = line.split(",");

        Text yearMonth = new Text(tokens[0] + tokens[1]);
        Text day = new Text(tokens[2]);

        final IntWritable temperature = new IntWritable(Integer.parseInt(tokens[3]));

        final DateTemperaturePair reduceKey = new DateTemperaturePair();
        reduceKey.setYearMonth(yearMonth);
        reduceKey.setDay(day);
        reduceKey.setTemperature(temperature);

        context.write(reduceKey, temperature);
    }
}
